package com.cavy.platform.server.service;

import com.cavy.platform.server.domain.job.ContainerLogSubscribeKey;
import com.cavy.platform.server.domain.job.ContainerLogUnSubscribeKey;
import com.cavy.platform.server.domain.job.Job;
import com.cavy.platform.server.domain.job.JobAction;
import com.cavy.platform.server.dto.JobSubscribeKey;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.websocket.Session;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;

@Service
@Slf4j
public class JobSubscribe {
    private ConcurrentHashMap<JobSubscribeKey, List<Session>> jobActionSubscribe = new ConcurrentHashMap<>();
    private ConcurrentHashMap<ContainerLogSubscribeKey, List<Session>> containerLogSubscribe = new ConcurrentHashMap<>();

    @Autowired
    private ContainerLogTailerManager tailerManager;
    @Autowired
    private JobService jobService;

    public void subscribe(Session session, String workSpace, String project, String pipeline, String number) {
        JobSubscribeKey key = new JobSubscribeKey(workSpace, project, pipeline, number);
        List<Session> list = this.jobActionSubscribe.get(key);
        if (list == null || list.isEmpty()) {
            list = new ArrayList<>();
        }
        log.debug("添加job:{}订阅", key);
        list.add(session);
        this.jobActionSubscribe.put(key, list);
    }

    public void unSubscribe(Session session, String workSpace, String project, String pipeline, String number) {
        JobSubscribeKey key = new JobSubscribeKey(workSpace, project, pipeline, number);
        List<Session> list = this.jobActionSubscribe.get(key);
        if (list == null || list.isEmpty()) {
            return;
        }
        log.debug("移除job:{}订阅", key);
        list.remove(session);
        this.jobActionSubscribe.put(key, list);
    }

    public void send(Job job, JobAction action) {
        JobSubscribeKey key = new JobSubscribeKey(job.getWorkspaceName(), job.getProjectName(), job.getPipelineName(),
                job.getNumber());
        List<Session> list = this.jobActionSubscribe.get(key);
        if (list == null || list.isEmpty()) {
            return;
        }
        for (Session session : list) {
            send(action, session);
        }
    }

    public void send(JobAction action, Session session) {
        try {
            log.debug("发送job状态:{}", action);
            session.getBasicRemote().sendText(action.toJson());
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    public void closeSession(Job job) {
        //关闭日志订阅
        closeSubscribe(job);
        //关闭session
        JobSubscribeKey key = new JobSubscribeKey(job.getWorkspaceName(), job.getProjectName(), job.getPipelineName(), job.getNumber());
        List<Session> list = this.jobActionSubscribe.get(key);
        if (list == null || list.isEmpty()) {
            return;
        }
        Iterator<Session> iterator = list.iterator();
        while (iterator.hasNext()) {
            Session next = iterator.next();
            iterator.remove();
            try {
                next.close();
            } catch (IOException e) {
                log.error("关闭客户端出现异常,异常类型:{},异常消息:{}", e.getClass(), e.getMessage());
                e.printStackTrace();
            }
        }
        this.jobActionSubscribe.remove(key);
    }

    private void closeSubscribe(Job job) {
        String workspaceName = job.getWorkspaceName();
        String projectName = job.getProjectName();
        String pipelineName = job.getPipelineName();
        String number = job.getNumber();
        Set<ContainerLogSubscribeKey> collect = containerLogSubscribe.keySet().stream().filter(new Predicate<ContainerLogSubscribeKey>() {
            @Override
            public boolean test(ContainerLogSubscribeKey key) {
                return key.getWorkSpace().equals(workspaceName) &&
                        key.getProject().equals(projectName) &&
                        key.getPipeline().equals(pipelineName) &&
                        key.getNumber().equals(number);
            }
        }).map(new Function<ContainerLogSubscribeKey, ContainerLogSubscribeKey>() {
            @Override
            public ContainerLogSubscribeKey apply(ContainerLogSubscribeKey key) {
                tailerManager.destroy(new ContainerLogUnSubscribeKey(key));
                return key;
            }
        }).collect(Collectors.toSet());
        containerLogSubscribe.keySet().removeAll(collect);
    }

    //订阅某步骤、取消订阅某步骤
    public void onMessage(Session session, String message, String workSpace, String project, String pipeline, String number) {
        ObjectMapper mapper = new ObjectMapper();
        ContainerLogSubscribeKey key;
        try {
            key = mapper.readValue(message, ContainerLogSubscribeKey.class);
        } catch (IOException e) {
            log.error("发生错误，错误类型:{},错误信息:{}", e.getClass().getName(), e.getMessage());
            throw new RuntimeException(e);
        }
        key.setWorkSpace(workSpace);
        key.setProject(project);
        key.setPipeline(pipeline);
        key.setNumber(number);
        key.handler(this.containerLogSubscribe, session, tailerManager);
    }

    public void send(String message, ContainerLogSubscribeKey key) {
        String convert = jobService.convert(message);
        containerLogSubscribe.get(key).forEach(new Consumer<Session>() {
            @Override
            public void accept(Session session) {
                try {
                    session.getBasicRemote().sendText(convert);
                } catch (IOException e) {
                    log.error("发生错误，错误类型:{},错误信息:{}", e.getClass().getName(), e.getMessage());
                    throw new RuntimeException(e);
                }
            }
        });
    }
}
